package com.lansetech.framework.redis.peak.shaving;

import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import redis.clients.util.SafeEncoder;

import java.util.*;
import java.util.concurrent.CountDownLatch;

public class PeakShavingListener {

    public static final String PEAKING_SHAVING_THREAD_PREFIX = "peaking-shaving-thread-";

    private final PeakShavingConfig config;

    private final PeakShavingRunnableContext context;

    private final PeakingShavingPatrol patrol;

    private final MessageConsumer consumer;

    private ConsumerGroup consumerGroup;

    public PeakShavingListener(JedisConnectionFactory factory) {
        this(factory, new DefaultMessageConsumer());
    }

    public PeakShavingListener(JedisConnectionFactory factory, MessageConsumer consumer) {
        this(factory, new PeakShavingConfig(), consumer);
    }

    public PeakShavingListener(JedisConnectionFactory factory, MessageConsumer consumer, int numberOfQueue) {
        this(factory, new PeakShavingConfig(numberOfQueue), consumer);
    }

    public PeakShavingListener(JedisConnectionFactory factory, PeakShavingConfig config) {
        this(factory, config, new DefaultMessageConsumer());
    }

    public PeakShavingListener(JedisConnectionFactory factory, PeakShavingConfig config, MessageConsumer consumer) {
        this.config = config;
        this.context = new PeakShavingRunnableContext(factory, config);
        this.patrol = new PeakingShavingPatrol(context);
        this.consumer = consumer;
        final PeakShavingListener listener = this;
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                listener.shutdown();
            }
        }));
    }

    public void init() {
        int queues = config.getNumberOfQueue();
        List<Thread> threads = new LinkedList<>();
        List<PeakShavingRunnable> runners = new LinkedList<>();
        for (int i = 0; i < queues; i++) {
            PeakShavingRunnable runner = createRunnable(i);
            runners.add(runner);
            threads.add(new Thread(runner, PEAKING_SHAVING_THREAD_PREFIX + i));
        }
        this.consumerGroup = new ConsumerGroup(context, threads);
        int latch = queues + 1;
        this.context.setShutdownLatch(new CountDownLatch(latch));
        patrol.registerQueues(runners);
        patrol.refreshQueues();
    }

    public void start() {
        patrol.start();
        this.consumerGroup.start();
    }

    public void shutdown() {
        this.consumerGroup.shutdown();
        try {
            this.context.getShutdownLatch().await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public PeakShavingRunnableContext getContext() {
        return context;
    }

    private PeakShavingRunnable createRunnable(int index) {
        PeakShavingRunnable runnable = new PeakShavingRunnable(this.patrol, context, config.getQueuePrefix(), index, consumer);
        context.registerConsumer(index);
        return runnable;
    }

    public Map<String, Integer> messageCount() {

        MessageQueueCountDTO count = new MessageQueueCountDTO();
        Map<String, Integer> normalMap = new HashMap<>();
        final Map<String, Integer> indexMap = new HashMap<>();
        final Map<String, String> topicMap = new HashMap<>();
        RedisConnection connection = null;

        try {
            connection = context.getConnectionFactory().getConnection();
            String prefix = context.getConfig().getQueuePrefix();
            Set<byte[]> kbs = connection.keys(SafeEncoder.encode(prefix + ":" + PeakShavingConfig.PEAK_SHAVING_TAG_NORMAL + ":*:*"));
            List<String> keys = new LinkedList<>();
            for (byte[] keyBytes : kbs) {
                keys.add(SafeEncoder.encode(keyBytes));
            }
            for (String key : keys) {
                String[] segments = decompose(key);
                if (segments == null) {
                    continue;
                }
                Long len = connection.lLen(SafeEncoder.encode(key));
                if (len == null || len == 0) {
                    continue;
                }
                String mt = segments[0];
                String type = segments[1];
                int index = Integer.parseInt(segments[2]);
                indexMap.put(key, index);
                topicMap.put(key, type);
                normalMap.put(key, len.intValue());
            }
        } finally {
            if (connection != null) {
                connection.close();
            }
        }


        List<String> normalKeys = new ArrayList<>(normalMap.keySet());

        Comparator<String> comparator = new Comparator<String>() {
            @Override
            public int compare(String o1, String o2) {
                String topic1 = topicMap.get(o1);
                String topic2 = topicMap.get(o2);
                int i = topic1.compareTo(topic2);
                if (i != 0) {
                    return i;
                }
                Integer index1 = indexMap.get(o1);
                Integer index2 = indexMap.get(o2);
                return index1.compareTo(index2);
            }
        };

        Collections.sort(normalKeys, comparator);

        LinkedHashMap<String, Integer> normalCountMap = new LinkedHashMap<>();

        for (String normalKey : normalKeys) {
            normalCountMap.put(normalKey, normalMap.get(normalKey));
        }

        return normalCountMap;
    }

    private String[] decompose(String key) {
        String[] split = key.split(":");
        if (split.length < 4) {
            return null;
        }
        String mt = split[2];
        if (!"normal".equals(mt)) {
            return null;
        }
        String type = split[3];
        if (type == null || type.trim().equals("")) {
            return null;
        }
        String index = split[4];
        try {
            Integer.parseInt(index);
        } catch (NumberFormatException e) {
            return null;
        }
        return new String[]{mt, type, index};
    }
}
